package com.qit.prototype.flink.performance;

import java.util.List;
import java.util.Map;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class TestCase1 {
	
	public TestCase1() throws Exception {
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
		
		DataStreamSource<String> dataStreamSource = env
				.socketTextStream("localhost", 11002, "\0");
		
		Pattern<String, ?> pattern = Pattern.<String>begin("start").where(new SimpleCondition<String>() {
			private static final long serialVersionUID = 1L;

			@Override
			public boolean filter(String evt) throws Exception {
//				return true;
				if (evt.lastIndexOf("|100000") >= 0) {
					return true;
				}
				return false;
			}
		});
		
		PatternStream<String> patternStream = CEP.pattern(dataStreamSource, pattern);
		
		patternStream.process(new PatternProcessFunction<String, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public void processMatch(
					Map<String, List<String>> arg0,
					org.apache.flink.cep.functions.PatternProcessFunction.Context arg1,
					Collector<String> arg2) throws Exception {
				
				System.out.println(System.currentTimeMillis());
				System.out.println(arg0);
				
			}
			
		});
		
        env.execute("Performance");
	}

}
